package realtime.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import realtime.util.MyKafkaUtil;

/**
 * @author MengX
 * @create 2023/3/25 10:02:53
 */
public class DwdVipChangeDetail {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String groupId = "vip_change_detail";
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL(groupId));

        Table vipChangeDetail = tableEnv.sqlQuery("" +
                "select \n" +
                "    data['id'] id,\n" +
                "    data['user_id'] user_id,\n" +
                "    data['from_vip'] from_vip,\n" +
                "    data['to_vip'] to_vip,\n" +
                "    data['create_time'] create_time\n" +
                "from `topic_db`\n" +
                "where `table` = 'vip_change_detail'  and  `type` = 'insert'");
        tableEnv.createTemporaryView("vip_change_detail",vipChangeDetail);

        tableEnv.executeSql("" +
                "CREATE TABLE `dwd_vip_change_detail` (\n" +
                "  `id`STRING,\n" +
                "  `user_id` STRING,\n" +
                "  `from_vip` STRING,\n" +
                "  `to_vip` STRING,\n" +
                "  `create_time` STRING\n" +
                ")\n" + MyKafkaUtil.getKafkaSinkDDL("dwd_vip_change_detail"));

        tableEnv.executeSql("insert into dwd_vip_change_detail select * from vip_change_detail");
    }
}
